package com.stone.framework.batch.tasklet;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;

/**
 * 文件拆分 tasklet
 * 
 * @author ybliu
 * @date 2020/07/27
 */
public class FileSplitingTasklet implements Tasklet, InitializingBean {

    // private static final Logger LOGGER = LoggerFactory.getLogger(FileSplitingTasklet.class);

    private Resource source;
    private String target;
    private int num;
    private Charset charset;
    private String placeholder = "INDEX";

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        if (!this.source.exists()) {
            throw new FileNotFoundException("文件不存在: " + this.source.getDescription());
        }
        List<Path> splitedPath = this.createSplitedFiles();
        Map<Integer, List<String>> splitedData = this.createSplitedData();
        this.writeToSplitedFiles(splitedData, splitedPath);

        contribution.setExitStatus(ExitStatus.COMPLETED);
        return RepeatStatus.FINISHED;
    }

    /**
     * 创建拆分文件
     * 
     * @return
     */
    private List<Path> createSplitedFiles() {
        return IntStream.range(0, this.num).mapToObj(index -> {
            Resource resource = new FileSystemResource(this.target.replaceAll(this.placeholder, String.valueOf(index)));
            Path path = null;
            try {
                path = Paths.get(resource.getURI());
                Files.deleteIfExists(path);
                Files.createFile(path);
            } catch (IOException e) {
                throw new RuntimeException("创建拆分文件失败", e);
            }
            return path;
        }).collect(Collectors.toList());
    }

    /**
     * 创建拆分数据
     * 
     * @return
     * @throws IOException
     */
    private Map<Integer, List<String>> createSplitedData() {
        try (Stream<String> stream = Files.lines(this.source.getFile().toPath(), charset)) {
            final AtomicInteger index = new AtomicInteger();
            return stream.peek(line -> index.getAndIncrement())
                .collect(Collectors.groupingBy(line -> index.get() % this.num));
        } catch (IOException e) {
            throw new RuntimeException("创建拆分数据失败", e);
        }
    }

    /**
     * 写入拆分文件
     * 
     * @return
     * @throws IOException
     */
    private void writeToSplitedFiles(Map<Integer, List<String>> splitedData, List<Path> splitedPath) {
        IntStream.range(0, this.num).parallel()
            .filter(index -> splitedData.get(index) != null && !splitedData.get(index).isEmpty()).forEach(index -> {
                try {
                    Files.write(splitedPath.get(index), splitedData.get(index), this.charset, StandardOpenOption.WRITE,
                        StandardOpenOption.APPEND);
                } catch (IOException e) {
                    throw new RuntimeException("写入拆分文件失败", e);
                }
            });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(this.source, "source must not be null");
        Assert.notNull(this.target, "target must not be null");
        Assert.notNull(this.charset, "charset must not be null");
        Assert.notNull(this.num, "num must not be null");
    }

    public void setSource(Resource source) {
        this.source = source;
    }

    public void setTarget(String target) {
        this.target = target;
    }

    public void setNum(int num) {
        this.num = num;
    }

    public void setCharset(Charset charset) {
        this.charset = charset;
    }

    public void setPlaceholder(String placeholder) {
        this.placeholder = placeholder;
    }

}
